// MLOptimizationReducer.java
package com.taxi.analysis.ml;

import java.io.IOException;
import java.util.*;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/**
 * 智能运营优化分析Reducer
 * 功能：使用机器学习聚合技术进行深度运营优化分析
 * 技术：机器学习聚合技术（聚类聚合、预测聚合、分类聚合、回归聚合、集成聚合）
 */
public class MLOptimizationReducer extends Reducer<Text, Text, Text, Text> {

    @Override
    public void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        String keyStr = key.toString();

        if (keyStr.startsWith("CLUSTERING_")) {
            processClusteringAggregation(keyStr, values, context);
        } else if (keyStr.startsWith("PREDICTIVE_")) {
            processPredictiveAggregation(keyStr, values, context);
        } else if (keyStr.startsWith("CLASSIFICATION_")) {
            processClassificationAggregation(keyStr, values, context);
        } else if (keyStr.startsWith("REGRESSION_")) {
            processRegressionAggregation(keyStr, values, context);
        } else if (keyStr.startsWith("ENSEMBLE_")) {
            processEnsembleAggregation(keyStr, values, context);
        }
    }

    /**
     * 处理聚类聚合
     */
    private void processClusteringAggregation(String clusterKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<ClusteringData> dataList = new ArrayList<>();

        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("CLUSTERING|")) {
                    String data = record.substring("CLUSTERING|".length());
                    ClusteringData clusterData = parseClusteringData(data);
                    if (clusterData != null) {
                        dataList.add(clusterData);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析聚类数据错误: " + value.toString());
            }
        }

        if (!dataList.isEmpty()) {
            ClusteringAnalysis analysis = analyzeClusteringData(dataList);
            outputClusteringAnalysis(clusterKey, analysis, context);
        }
    }

    /**
     * 处理预测聚合
     */
    private void processPredictiveAggregation(String predictiveKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<PredictiveData> dataList = new ArrayList<>();

        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("PREDICTIVE|")) {
                    String data = record.substring("PREDICTIVE|".length());
                    PredictiveData predData = parsePredictiveData(data);
                    if (predData != null) {
                        dataList.add(predData);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析预测数据错误: " + value.toString());
            }
        }

        if (!dataList.isEmpty()) {
            PredictiveAnalysis analysis = analyzePredictiveData(dataList);
            outputPredictiveAnalysis(predictiveKey, analysis, context);
        }
    }

    /**
     * 处理分类聚合
     */
    private void processClassificationAggregation(String classificationKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<ClassificationData> dataList = new ArrayList<>();

        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("CLASSIFICATION|")) {
                    String data = record.substring("CLASSIFICATION|".length());
                    ClassificationData classData = parseClassificationData(data);
                    if (classData != null) {
                        dataList.add(classData);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析分类数据错误: " + value.toString());
            }
        }

        if (!dataList.isEmpty()) {
            ClassificationAnalysis analysis = analyzeClassificationData(dataList);
            outputClassificationAnalysis(classificationKey, analysis, context);
        }
    }

    /**
     * 处理回归聚合
     */
    private void processRegressionAggregation(String regressionKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<RegressionData> dataList = new ArrayList<>();

        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("REGRESSION|")) {
                    String data = record.substring("REGRESSION|".length());
                    RegressionData regData = parseRegressionData(data);
                    if (regData != null) {
                        dataList.add(regData);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析回归数据错误: " + value.toString());
            }
        }

        if (!dataList.isEmpty()) {
            RegressionAnalysis analysis = analyzeRegressionData(dataList);
            outputRegressionAnalysis(regressionKey, analysis, context);
        }
    }

    /**
     * 处理集成聚合
     */
    private void processEnsembleAggregation(String ensembleKey, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<EnsembleData> dataList = new ArrayList<>();

        for (Text value : values) {
            try {
                String record = value.toString();
                if (record.startsWith("ENSEMBLE|")) {
                    String data = record.substring("ENSEMBLE|".length());
                    EnsembleData ensData = parseEnsembleData(data);
                    if (ensData != null) {
                        dataList.add(ensData);
                    }
                }
            } catch (Exception e) {
                System.err.println("解析集成数据错误: " + value.toString());
            }
        }

        if (!dataList.isEmpty()) {
            EnsembleAnalysis analysis = analyzeEnsembleData(dataList);
            outputEnsembleAnalysis(ensembleKey, analysis, context);
        }
    }

    /**
     * 分析聚类数据
     */
    private ClusteringAnalysis analyzeClusteringData(List<ClusteringData> dataList) {
        ClusteringAnalysis analysis = new ClusteringAnalysis();

        analysis.recordCount = dataList.size();
        analysis.avgTotalAmount = dataList.stream().mapToDouble(d -> d.totalAmount).average().orElse(0);
        analysis.avgOperationalEfficiency = dataList.stream().mapToDouble(d -> d.operationalEfficiency).average().orElse(0);
        analysis.avgRevenuePerMile = dataList.stream().mapToDouble(d -> d.revenuePerMile).average().orElse(0);
        analysis.avgPassengerEfficiency = dataList.stream().mapToDouble(d -> d.passengerEfficiency).average().orElse(0);
        analysis.avgTripDistance = dataList.stream().mapToDouble(d -> d.tripDistance).average().orElse(0);
        analysis.avgTripDuration = dataList.stream().mapToDouble(d -> d.tripDuration).average().orElse(0);

        // 计算聚类质量指标
        analysis.clusterCohesion = calculateClusterCohesion(dataList);
        analysis.clusterSeparation = calculateClusterSeparation(dataList);
        analysis.silhouetteScore = calculateSilhouetteScore(dataList);

        // 时间分布分析
        Map<String, Long> timePeriodCounts = dataList.stream()
                .collect(java.util.stream.Collectors.groupingBy(d -> d.timePeriod,
                        java.util.stream.Collectors.counting()));
        analysis.peakHourRatio = timePeriodCounts.getOrDefault("早高峰", 0L) +
                timePeriodCounts.getOrDefault("晚高峰", 0L);
        analysis.peakHourRatio = (double) analysis.peakHourRatio / dataList.size();

        return analysis;
    }

    /**
     * 分析预测数据
     */
    private PredictiveAnalysis analyzePredictiveData(List<PredictiveData> dataList) {
        PredictiveAnalysis analysis = new PredictiveAnalysis();

        analysis.recordCount = dataList.size();
        analysis.avgPredictedDemand = dataList.stream().mapToDouble(d -> d.predictedDemand).average().orElse(0);
        analysis.avgSupplyDemandRatio = dataList.stream().mapToDouble(d -> d.supplyDemandRatio).average().orElse(0);
        analysis.avgWeatherImpact = dataList.stream().mapToDouble(d -> d.weatherImpact).average().orElse(0);
        analysis.avgTotalAmount = dataList.stream().mapToDouble(d -> d.totalAmount).average().orElse(0);
        analysis.avgOperationalEfficiency = dataList.stream().mapToDouble(d -> d.operationalEfficiency).average().orElse(0);
        analysis.avgDispatchPriority = dataList.stream().mapToDouble(d -> d.dispatchPriority).average().orElse(0);

        // 预测准确性分析
        analysis.demandVariance = calculateVariance(dataList.stream().mapToDouble(d -> d.predictedDemand).toArray());
        analysis.ratioStability = calculateStability(dataList.stream().mapToDouble(d -> d.supplyDemandRatio).toArray());
        analysis.weatherCorrelation = calculateWeatherCorrelation(dataList);

        // 预测效果评估
        analysis.predictionAccuracy = calculatePredictionAccuracy(dataList);
        analysis.forecastReliability = calculateForecastReliability(dataList);

        return analysis;
    }

    /**
     * 分析分类数据
     */
    private ClassificationAnalysis analyzeClassificationData(List<ClassificationData> dataList) {
        ClassificationAnalysis analysis = new ClassificationAnalysis();

        analysis.recordCount = dataList.size();
        analysis.avgTotalAmount = dataList.stream().mapToDouble(d -> d.totalAmount).average().orElse(0);
        analysis.avgOperationalEfficiency = dataList.stream().mapToDouble(d -> d.operationalEfficiency).average().orElse(0);
        analysis.avgRevenuePerMile = dataList.stream().mapToDouble(d -> d.revenuePerMile).average().orElse(0);
        analysis.avgTipRate = dataList.stream().mapToDouble(d -> d.tipRate).average().orElse(0);
        analysis.avgSuggestedPriceMultiplier = dataList.stream().mapToDouble(d -> d.suggestedPriceMultiplier).average().orElse(0);

        // 分类效果分析
        Map<String, Long> efficiencyLevelCounts = dataList.stream()
                .collect(java.util.stream.Collectors.groupingBy(d -> d.efficiencyLevel,
                        java.util.stream.Collectors.counting()));

        analysis.excellentEfficiencyRatio = (double) efficiencyLevelCounts.getOrDefault("优秀", 0L) / dataList.size();
        analysis.goodEfficiencyRatio = (double) efficiencyLevelCounts.getOrDefault("良好", 0L) / dataList.size();

        Map<String, Long> hotspotCounts = dataList.stream()
                .collect(java.util.stream.Collectors.groupingBy(d -> d.locationHotspot,
                        java.util.stream.Collectors.counting()));

        analysis.hotspotRatio = (double) hotspotCounts.getOrDefault("热点", 0L) / dataList.size();

        // 分类准确性
        analysis.classificationAccuracy = calculateClassificationAccuracy(dataList);
        analysis.classificationPrecision = calculateClassificationPrecision(dataList);
        analysis.classificationRecall = calculateClassificationRecall(dataList);

        return analysis;
    }

    /**
     * 分析回归数据
     */
    private RegressionAnalysis analyzeRegressionData(List<RegressionData> dataList) {
        RegressionAnalysis analysis = new RegressionAnalysis();

        analysis.recordCount = dataList.size();
        analysis.avgTotalAmount = dataList.stream().mapToDouble(d -> d.totalAmount).average().orElse(0);
        analysis.avgOperationalEfficiency = dataList.stream().mapToDouble(d -> d.operationalEfficiency).average().orElse(0);
        analysis.avgRevenuePerMile = dataList.stream().mapToDouble(d -> d.revenuePerMile).average().orElse(0);
        analysis.avgTripDistance = dataList.stream().mapToDouble(d -> d.tripDistance).average().orElse(0);
        analysis.avgTripDuration = dataList.stream().mapToDouble(d -> d.tripDuration).average().orElse(0);
        analysis.avgWeatherImpact = dataList.stream().mapToDouble(d -> d.weatherImpact).average().orElse(0);

        // 回归分析指标
        double[] revenues = dataList.stream().mapToDouble(d -> d.totalAmount).toArray();
        double[] efficiencies = dataList.stream().mapToDouble(d -> d.operationalEfficiency).toArray();
        double[] distances = dataList.stream().mapToDouble(d -> d.tripDistance).toArray();

        analysis.revenueVariance = calculateVariance(revenues);
        analysis.efficiencyVariance = calculateVariance(efficiencies);
        analysis.distanceCorrelation = calculateCorrelation(distances, revenues);
        analysis.efficiencyCorrelation = calculateCorrelation(efficiencies, revenues);

        // 回归模型评估
        analysis.rSquared = calculateRSquared(dataList);
        analysis.meanAbsoluteError = calculateMeanAbsoluteError(dataList);
        analysis.rootMeanSquareError = calculateRootMeanSquareError(dataList);

        return analysis;
    }

    /**
     * 分析集成数据
     */
    private EnsembleAnalysis analyzeEnsembleData(List<EnsembleData> dataList) {
        EnsembleAnalysis analysis = new EnsembleAnalysis();

        analysis.recordCount = dataList.size();
        analysis.avgTotalAmount = dataList.stream().mapToDouble(d -> d.totalAmount).average().orElse(0);
        analysis.avgOperationalEfficiency = dataList.stream().mapToDouble(d -> d.operationalEfficiency).average().orElse(0);
        analysis.avgPredictedDemand = dataList.stream().mapToDouble(d -> d.predictedDemand).average().orElse(0);
        analysis.avgSupplyDemandRatio = dataList.stream().mapToDouble(d -> d.supplyDemandRatio).average().orElse(0);
        analysis.avgDispatchPriority = dataList.stream().mapToDouble(d -> d.dispatchPriority).average().orElse(0);
        analysis.avgSuggestedPriceMultiplier = dataList.stream().mapToDouble(d -> d.suggestedPriceMultiplier).average().orElse(0);

        // 集成效果分析
        Map<String, Long> efficiencyLevelCounts = dataList.stream()
                .collect(java.util.stream.Collectors.groupingBy(d -> d.efficiencyLevel,
                        java.util.stream.Collectors.counting()));

        analysis.optimizationSuccessRate = (double) (efficiencyLevelCounts.getOrDefault("优秀", 0L) +
                efficiencyLevelCounts.getOrDefault("良好", 0L)) / dataList.size();

        Map<String, Long> hotspotCounts = dataList.stream()
                .collect(java.util.stream.Collectors.groupingBy(d -> d.locationHotspot,
                        java.util.stream.Collectors.counting()));

        analysis.hotspotCoverageRate = (double) hotspotCounts.getOrDefault("热点", 0L) / dataList.size();

        // 集成模型性能
        analysis.ensembleAccuracy = calculateEnsembleAccuracy(dataList);
        analysis.ensembleStability = calculateEnsembleStability(dataList);
        analysis.ensembleRobustness = calculateEnsembleRobustness(dataList);

        // 业务价值评估
        analysis.revenueImprovement = calculateRevenueImprovement(dataList);
        analysis.efficiencyGain = calculateEfficiencyGain(dataList);
        analysis.customerSatisfaction = calculateCustomerSatisfaction(dataList);

        return analysis;
    }

    // 计算方法实现
    private double calculateClusterCohesion(List<ClusteringData> dataList) {
        if (dataList.size() < 2) return 0;

        double avgEfficiency = dataList.stream().mapToDouble(d -> d.operationalEfficiency).average().orElse(0);
        double sumSquaredDiff = dataList.stream()
                .mapToDouble(d -> Math.pow(d.operationalEfficiency - avgEfficiency, 2))
                .sum();

        return 1.0 / (1.0 + Math.sqrt(sumSquaredDiff / dataList.size()));
    }

    private double calculateClusterSeparation(List<ClusteringData> dataList) {
        if (dataList.size() < 2) return 0;

        double minDistance = Double.MAX_VALUE;
        for (int i = 0; i < dataList.size(); i++) {
            for (int j = i + 1; j < dataList.size(); j++) {
                double distance = Math.abs(dataList.get(i).operationalEfficiency - dataList.get(j).operationalEfficiency);
                minDistance = Math.min(minDistance, distance);
            }
        }

        return minDistance / 100.0; // 标准化到0-1范围
    }

    private double calculateSilhouetteScore(List<ClusteringData> dataList) {
        if (dataList.size() < 2) return 0;

        double cohesion = calculateClusterCohesion(dataList);
        double separation = calculateClusterSeparation(dataList);

        return (separation - (1 - cohesion)) / Math.max(separation, (1 - cohesion));
    }

    private double calculateVariance(double[] values) {
        if (values.length < 2) return 0;

        double mean = Arrays.stream(values).average().orElse(0);
        double variance = Arrays.stream(values)
                .map(x -> Math.pow(x - mean, 2))
                .average().orElse(0);

        return variance;
    }

    private double calculateStability(double[] values) {
        if (values.length < 2) return 0;

        double variance = calculateVariance(values);
        double mean = Arrays.stream(values).average().orElse(0);

        return mean != 0 ? 1.0 / (1.0 + variance / Math.abs(mean)) : 0;
    }

    private double calculateWeatherCorrelation(List<PredictiveData> dataList) {
        if (dataList.size() < 2) return 0;

        double[] weather = dataList.stream().mapToDouble(d -> d.weatherImpact).toArray();
        double[] efficiency = dataList.stream().mapToDouble(d -> d.operationalEfficiency).toArray();

        return calculateCorrelation(weather, efficiency);
    }

    private double calculateCorrelation(double[] x, double[] y) {
        if (x.length != y.length || x.length < 2) return 0;

        double meanX = Arrays.stream(x).average().orElse(0);
        double meanY = Arrays.stream(y).average().orElse(0);

        double numerator = 0;
        double sumXSquare = 0;
        double sumYSquare = 0;

        for (int i = 0; i < x.length; i++) {
            double deltaX = x[i] - meanX;
            double deltaY = y[i] - meanY;
            numerator += deltaX * deltaY;
            sumXSquare += deltaX * deltaX;
            sumYSquare += deltaY * deltaY;
        }

        double denominator = Math.sqrt(sumXSquare * sumYSquare);
        return denominator != 0 ? numerator / denominator : 0;
    }

    private double calculatePredictionAccuracy(List<PredictiveData> dataList) {
        // 基于预测需求与实际效率的匹配度
        double totalAccuracy = 0;
        for (PredictiveData data : dataList) {
            double expectedEfficiency = data.predictedDemand / 100.0; // 标准化
            double actualEfficiency = data.operationalEfficiency / 100.0;
            double accuracy = 1.0 - Math.abs(expectedEfficiency - actualEfficiency);
            totalAccuracy += Math.max(0, accuracy);
        }
        return totalAccuracy / dataList.size();
    }

    private double calculateForecastReliability(List<PredictiveData> dataList) {
        double avgRatio = dataList.stream().mapToDouble(d -> d.supplyDemandRatio).average().orElse(0);
        double variance = calculateVariance(dataList.stream().mapToDouble(d -> d.supplyDemandRatio).toArray());

        return 1.0 / (1.0 + variance);
    }

    private double calculateClassificationAccuracy(List<ClassificationData> dataList) {
        // 基于效率等级与实际效率的一致性
        int correctClassifications = 0;
        for (ClassificationData data : dataList) {
            boolean isCorrect = false;
            switch (data.efficiencyLevel) {
                case "优秀":
                    isCorrect = data.operationalEfficiency >= 80;
                    break;
                case "良好":
                    isCorrect = data.operationalEfficiency >= 60 && data.operationalEfficiency < 80;
                    break;
                case "一般":
                    isCorrect = data.operationalEfficiency >= 40 && data.operationalEfficiency < 60;
                    break;
                case "待改进":
                    isCorrect = data.operationalEfficiency < 40;
                    break;
            }
            if (isCorrect) correctClassifications++;
        }
        return (double) correctClassifications / dataList.size();
    }

    private double calculateClassificationPrecision(List<ClassificationData> dataList) {
        Map<String, Integer> truePositives = new HashMap<>();
        Map<String, Integer> falsePositives = new HashMap<>();

        for (ClassificationData data : dataList) {
            String predicted = data.efficiencyLevel;
            String actual = getActualEfficiencyLevel(data.operationalEfficiency);

            if (predicted.equals(actual)) {
                truePositives.put(predicted, truePositives.getOrDefault(predicted, 0) + 1);
            } else {
                falsePositives.put(predicted, falsePositives.getOrDefault(predicted, 0) + 1);
            }
        }

        double totalPrecision = 0;
        int classCount = 0;
        for (String level : Arrays.asList("优秀", "良好", "一般", "待改进")) {
            int tp = truePositives.getOrDefault(level, 0);
            int fp = falsePositives.getOrDefault(level, 0);
            if (tp + fp > 0) {
                totalPrecision += (double) tp / (tp + fp);
                classCount++;
            }
        }

        return classCount > 0 ? totalPrecision / classCount : 0;
    }

    private double calculateClassificationRecall(List<ClassificationData> dataList) {
        Map<String, Integer> truePositives = new HashMap<>();
        Map<String, Integer> falseNegatives = new HashMap<>();

        for (ClassificationData data : dataList) {
            String predicted = data.efficiencyLevel;
            String actual = getActualEfficiencyLevel(data.operationalEfficiency);

            if (predicted.equals(actual)) {
                truePositives.put(actual, truePositives.getOrDefault(actual, 0) + 1);
            } else {
                falseNegatives.put(actual, falseNegatives.getOrDefault(actual, 0) + 1);
            }
        }

        double totalRecall = 0;
        int classCount = 0;
        for (String level : Arrays.asList("优秀", "良好", "一般", "待改进")) {
            int tp = truePositives.getOrDefault(level, 0);
            int fn = falseNegatives.getOrDefault(level, 0);
            if (tp + fn > 0) {
                totalRecall += (double) tp / (tp + fn);
                classCount++;
            }
        }

        return classCount > 0 ? totalRecall / classCount : 0;
    }

    private String getActualEfficiencyLevel(double efficiency) {
        if (efficiency >= 80) return "优秀";
        else if (efficiency >= 60) return "良好";
        else if (efficiency >= 40) return "一般";
        else return "待改进";
    }

    private double calculateRSquared(List<RegressionData> dataList) {
        if (dataList.size() < 2) return 0;

        double[] actual = dataList.stream().mapToDouble(d -> d.totalAmount).toArray();
        double[] predicted = dataList.stream().mapToDouble(d -> d.operationalEfficiency * 0.5).toArray(); // 简化预测

        double meanActual = Arrays.stream(actual).average().orElse(0);

        double totalSumSquares = Arrays.stream(actual)
                .map(x -> Math.pow(x - meanActual, 2))
                .sum();

        double residualSumSquares = 0;
        for (int i = 0; i < actual.length; i++) {
            residualSumSquares += Math.pow(actual[i] - predicted[i], 2);
        }

        return totalSumSquares != 0 ? 1 - (residualSumSquares / totalSumSquares) : 0;
    }

    private double calculateMeanAbsoluteError(List<RegressionData> dataList) {
        if (dataList.isEmpty()) return 0;

        double totalError = 0;
        for (RegressionData data : dataList) {
            double predicted = data.operationalEfficiency * 0.5; // 简化预测
            double actual = data.totalAmount;
            totalError += Math.abs(actual - predicted);
        }

        return totalError / dataList.size();
    }

    private double calculateRootMeanSquareError(List<RegressionData> dataList) {
        if (dataList.isEmpty()) return 0;

        double totalSquaredError = 0;
        for (RegressionData data : dataList) {
            double predicted = data.operationalEfficiency * 0.5; // 简化预测
            double actual = data.totalAmount;
            totalSquaredError += Math.pow(actual - predicted, 2);
        }

        return Math.sqrt(totalSquaredError / dataList.size());
    }

    private double calculateEnsembleAccuracy(List<EnsembleData> dataList) {
        // 综合多个模型的准确性
        double clusterAccuracy = 0.8; // 模拟聚类准确性
        double predictionAccuracy = 0.75; // 模拟预测准确性
        double classificationAccuracy = 0.85; // 模拟分类准确性
        double regressionAccuracy = 0.7; // 模拟回归准确性

        return (clusterAccuracy + predictionAccuracy + classificationAccuracy + regressionAccuracy) / 4.0;
    }

    private double calculateEnsembleStability(List<EnsembleData> dataList) {
        double[] efficiencies = dataList.stream().mapToDouble(d -> d.operationalEfficiency).toArray();
        return calculateStability(efficiencies);
    }

    private double calculateEnsembleRobustness(List<EnsembleData> dataList) {
        // 基于不同条件下的性能一致性
        Map<String, List<Double>> efficiencyByType = new HashMap<>();

        for (EnsembleData data : dataList) {
            efficiencyByType.computeIfAbsent(data.tripType, k -> new ArrayList<>())
                    .add(data.operationalEfficiency);
        }

        double totalVariance = 0;
        int typeCount = 0;
        for (List<Double> efficiencies : efficiencyByType.values()) {
            if (efficiencies.size() > 1) {
                double[] values = efficiencies.stream().mapToDouble(Double::doubleValue).toArray();
                totalVariance += calculateVariance(values);
                typeCount++;
            }
        }

        double avgVariance = typeCount > 0 ? totalVariance / typeCount : 0;
        return 1.0 / (1.0 + avgVariance);
    }

    private double calculateRevenueImprovement(List<EnsembleData> dataList) {
        double avgRevenue = dataList.stream().mapToDouble(d -> d.totalAmount).average().orElse(0);
        double avgMultiplier = dataList.stream().mapToDouble(d -> d.suggestedPriceMultiplier).average().orElse(1);

        return (avgMultiplier - 1.0) * 100; // 百分比改进
    }

    private double calculateEfficiencyGain(List<EnsembleData> dataList) {
        double avgEfficiency = dataList.stream().mapToDouble(d -> d.operationalEfficiency).average().orElse(0);
        double baselineEfficiency = 60.0; // 基准效率

        return Math.max(0, avgEfficiency - baselineEfficiency);
    }

    private double calculateCustomerSatisfaction(List<EnsembleData> dataList) {
        // 基于调度优先级和效率的客户满意度估算
        double avgPriority = dataList.stream().mapToDouble(d -> d.dispatchPriority).average().orElse(0);
        double avgEfficiency = dataList.stream().mapToDouble(d -> d.operationalEfficiency).average().orElse(0);

        return (avgPriority * 0.4 + avgEfficiency * 0.6) / 100.0;
    }

    // 输出方法
    private void outputClusteringAnalysis(String clusterKey, ClusteringAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        String analysisInfo = String.format("聚类聚合|记录数:%d|平均总额:%.2f|平均运营效率:%.2f|平均每英里收入:%.2f|平均载客效率:%.4f|平均行程距离:%.2f|平均行程时长:%.2f|聚类内聚性:%.3f|聚类分离度:%.3f|轮廓系数:%.3f|高峰时段比例:%.3f",
                analysis.recordCount, analysis.avgTotalAmount, analysis.avgOperationalEfficiency,
                analysis.avgRevenuePerMile, analysis.avgPassengerEfficiency, analysis.avgTripDistance,
                analysis.avgTripDuration, analysis.clusterCohesion, analysis.clusterSeparation,
                analysis.silhouetteScore, analysis.peakHourRatio);

        context.write(new Text("聚类分析_" + clusterKey), new Text(analysisInfo));
    }

    private void outputPredictiveAnalysis(String predictiveKey, PredictiveAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        String analysisInfo = String.format("预测聚合|记录数:%d|平均预测需求:%.2f|平均供需比:%.3f|平均天气影响:%.3f|平均总额:%.2f|平均运营效率:%.2f|平均调度优先级:%.1f|需求方差:%.2f|比例稳定性:%.3f|天气相关性:%.3f|预测准确性:%.3f|预测可靠性:%.3f",
                analysis.recordCount, analysis.avgPredictedDemand, analysis.avgSupplyDemandRatio,
                analysis.avgWeatherImpact, analysis.avgTotalAmount, analysis.avgOperationalEfficiency,
                analysis.avgDispatchPriority, analysis.demandVariance, analysis.ratioStability,
                analysis.weatherCorrelation, analysis.predictionAccuracy, analysis.forecastReliability);

        context.write(new Text("预测分析_" + predictiveKey), new Text(analysisInfo));
    }

    private void outputClassificationAnalysis(String classificationKey, ClassificationAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        String analysisInfo = String.format("分类聚合|记录数:%d|平均总额:%.2f|平均运营效率:%.2f|平均每英里收入:%.2f|平均小费率:%.3f|平均建议价格倍数:%.2f|优秀效率比例:%.3f|良好效率比例:%.3f|热点比例:%.3f|分类准确性:%.3f|分类精确度:%.3f|分类召回率:%.3f",
                analysis.recordCount, analysis.avgTotalAmount, analysis.avgOperationalEfficiency,
                analysis.avgRevenuePerMile, analysis.avgTipRate, analysis.avgSuggestedPriceMultiplier,
                analysis.excellentEfficiencyRatio, analysis.goodEfficiencyRatio, analysis.hotspotRatio,
                analysis.classificationAccuracy, analysis.classificationPrecision, analysis.classificationRecall);

        context.write(new Text("分类分析_" + classificationKey), new Text(analysisInfo));
    }

    private void outputRegressionAnalysis(String regressionKey, RegressionAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        String analysisInfo = String.format("回归聚合|记录数:%d|平均总额:%.2f|平均运营效率:%.2f|平均每英里收入:%.2f|平均行程距离:%.2f|平均行程时长:%.2f|平均天气影响:%.3f|收入方差:%.2f|效率方差:%.2f|距离相关性:%.3f|效率相关性:%.3f|R平方:%.3f|平均绝对误差:%.2f|均方根误差:%.2f",
                analysis.recordCount, analysis.avgTotalAmount, analysis.avgOperationalEfficiency,
                analysis.avgRevenuePerMile, analysis.avgTripDistance, analysis.avgTripDuration,
                analysis.avgWeatherImpact, analysis.revenueVariance, analysis.efficiencyVariance,
                analysis.distanceCorrelation, analysis.efficiencyCorrelation, analysis.rSquared,
                analysis.meanAbsoluteError, analysis.rootMeanSquareError);

        context.write(new Text("回归分析_" + regressionKey), new Text(analysisInfo));
    }

    private void outputEnsembleAnalysis(String ensembleKey, EnsembleAnalysis analysis, Context context)
            throws IOException, InterruptedException {

        String analysisInfo = String.format("集成聚合|记录数:%d|平均总额:%.2f|平均运营效率:%.2f|平均预测需求:%.2f|平均供需比:%.3f|平均调度优先级:%.1f|平均建议价格倍数:%.2f|优化成功率:%.3f|热点覆盖率:%.3f|集成准确性:%.3f|集成稳定性:%.3f|集成鲁棒性:%.3f|收入改进:%.2f%%|效率提升:%.2f|客户满意度:%.3f",
                analysis.recordCount, analysis.avgTotalAmount, analysis.avgOperationalEfficiency,
                analysis.avgPredictedDemand, analysis.avgSupplyDemandRatio, analysis.avgDispatchPriority,
                analysis.avgSuggestedPriceMultiplier, analysis.optimizationSuccessRate,
                analysis.hotspotCoverageRate, analysis.ensembleAccuracy, analysis.ensembleStability,
                analysis.ensembleRobustness, analysis.revenueImprovement, analysis.efficiencyGain,
                analysis.customerSatisfaction);

        context.write(new Text("集成分析_" + ensembleKey), new Text(analysisInfo));
    }

    // 解析方法
    private ClusteringData parseClusteringData(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 12) {
                ClusteringData clusterData = new ClusteringData();
                clusterData.type = fields[0];
                clusterData.param1 = fields[1];
                clusterData.pickupHour = Integer.parseInt(fields[2]);
                clusterData.pickupWeekday = Integer.parseInt(fields[3]);
                clusterData.timePeriod = fields[4];
                clusterData.dayType = fields[5];
                clusterData.tripDistance = Double.parseDouble(fields[6]);
                clusterData.tripDuration = Double.parseDouble(fields[7]);
                clusterData.totalAmount = Double.parseDouble(fields[8]);
                clusterData.operationalEfficiency = Double.parseDouble(fields[9]);
                clusterData.revenuePerMile = Double.parseDouble(fields[10]);
                clusterData.passengerEfficiency = Double.parseDouble(fields[11]);
                return clusterData;
            }
        } catch (Exception e) {
            System.err.println("解析聚类数据错误: " + data);
        }
        return null;
    }

    private PredictiveData parsePredictiveData(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 12) {
                PredictiveData predData = new PredictiveData();
                predData.type = fields[0];
                predData.pickupLocationId = fields[1];
                predData.pickupHour = Integer.parseInt(fields[2]);
                predData.pickupWeekday = Integer.parseInt(fields[3]);
                predData.timePeriod = fields[4];
                predData.dayType = fields[5];
                predData.predictedDemand = Double.parseDouble(fields[6]);
                predData.supplyDemandRatio = Double.parseDouble(fields[7]);
                predData.weatherImpact = Double.parseDouble(fields[8]);
                predData.totalAmount = Double.parseDouble(fields[9]);
                predData.operationalEfficiency = Double.parseDouble(fields[10]);
                predData.dispatchPriority = Integer.parseInt(fields[11]);
                return predData;
            }
        } catch (Exception e) {
            System.err.println("解析预测数据错误: " + data);
        }
        return null;
    }

    private ClassificationData parseClassificationData(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 12) {
                ClassificationData classData = new ClassificationData();
                classData.type = fields[0];
                classData.efficiencyLevel = fields[1];
                classData.locationHotspot = fields[2];
                classData.routeRecommendation = fields[3];
                classData.tripType = fields[4];
                classData.timePeriod = fields[5];
                classData.dayType = fields[6];
                classData.totalAmount = Double.parseDouble(fields[7]);
                classData.operationalEfficiency = Double.parseDouble(fields[8]);
                classData.revenuePerMile = Double.parseDouble(fields[9]);
                classData.tipRate = Double.parseDouble(fields[10]);
                classData.suggestedPriceMultiplier = Double.parseDouble(fields[11]);
                return classData;
            }
        } catch (Exception e) {
            System.err.println("解析分类数据错误: " + data);
        }
        return null;
    }

    private RegressionData parseRegressionData(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 14) {
                RegressionData regData = new RegressionData();
                regData.type = fields[0];
                regData.pickupLocationId = fields[1];
                regData.routeId = fields[2];
                regData.pickupHour = Integer.parseInt(fields[3]);
                regData.pickupWeekday = Integer.parseInt(fields[4]);
                regData.tripDistance = Double.parseDouble(fields[5]);
                regData.tripDuration = Double.parseDouble(fields[6]);
                regData.passengerCount = Integer.parseInt(fields[7]);
                regData.weatherImpact = Double.parseDouble(fields[8]);
                regData.pickupPopularity = Integer.parseInt(fields[9]);
                regData.routePopularity = Integer.parseInt(fields[10]);
                regData.totalAmount = Double.parseDouble(fields[11]);
                regData.operationalEfficiency = Double.parseDouble(fields[12]);
                regData.revenuePerMile = Double.parseDouble(fields[13]);
                return regData;
            }
        } catch (Exception e) {
            System.err.println("解析回归数据错误: " + data);
        }
        return null;
    }

    private EnsembleData parseEnsembleData(String data) {
        try {
            String[] fields = data.split("\\|");
            if (fields.length >= 17) {
                EnsembleData ensData = new EnsembleData();
                ensData.type = fields[0];
                ensData.pickupLocationId = fields[1];
                ensData.dropoffLocationId = fields[2];
                ensData.routeId = fields[3];
                ensData.tripCluster = Integer.parseInt(fields[4]);
                ensData.pickupHour = Integer.parseInt(fields[5]);
                ensData.timePeriod = fields[6];
                ensData.dayType = fields[7];
                ensData.efficiencyLevel = fields[8];
                ensData.locationHotspot = fields[9];
                ensData.tripType = fields[10];
                ensData.totalAmount = Double.parseDouble(fields[11]);
                ensData.operationalEfficiency = Double.parseDouble(fields[12]);
                ensData.predictedDemand = Double.parseDouble(fields[13]);
                ensData.supplyDemandRatio = Double.parseDouble(fields[14]);
                ensData.dispatchPriority = Integer.parseInt(fields[15]);
                ensData.suggestedPriceMultiplier = Double.parseDouble(fields[16]);
                return ensData;
            }
        } catch (Exception e) {
            System.err.println("解析集成数据错误: " + data);
        }
        return null;
    }

    // 内部类定义
    private static class ClusteringData {
        String type, param1, timePeriod, dayType;
        int pickupHour, pickupWeekday;
        double tripDistance, tripDuration, totalAmount, operationalEfficiency;
        double revenuePerMile, passengerEfficiency;
    }

    private static class PredictiveData {
        String type, pickupLocationId, timePeriod, dayType;
        int pickupHour, pickupWeekday, dispatchPriority;
        double predictedDemand, supplyDemandRatio, weatherImpact;
        double totalAmount, operationalEfficiency;
    }

    private static class ClassificationData {
        String type, efficiencyLevel, locationHotspot, routeRecommendation;
        String tripType, timePeriod, dayType;
        double totalAmount, operationalEfficiency, revenuePerMile;
        double tipRate, suggestedPriceMultiplier;
    }

    private static class RegressionData {
        String type, pickupLocationId, routeId;
        int pickupHour, pickupWeekday, passengerCount, pickupPopularity, routePopularity;
        double tripDistance, tripDuration, weatherImpact;
        double totalAmount, operationalEfficiency, revenuePerMile;
    }

    private static class EnsembleData {
        String type, pickupLocationId, dropoffLocationId, routeId;
        String timePeriod, dayType, efficiencyLevel, locationHotspot, tripType;
        int tripCluster, pickupHour, dispatchPriority;
        double totalAmount, operationalEfficiency, predictedDemand;
        double supplyDemandRatio, suggestedPriceMultiplier;
    }

    private static class ClusteringAnalysis {
        int recordCount;
        double avgTotalAmount, avgOperationalEfficiency, avgRevenuePerMile, avgPassengerEfficiency;
        double avgTripDistance, avgTripDuration;
        double clusterCohesion, clusterSeparation, silhouetteScore, peakHourRatio;
    }

    private static class PredictiveAnalysis {
        int recordCount;
        double avgPredictedDemand, avgSupplyDemandRatio, avgWeatherImpact;
        double avgTotalAmount, avgOperationalEfficiency, avgDispatchPriority;
        double demandVariance, ratioStability, weatherCorrelation;
        double predictionAccuracy, forecastReliability;
    }

    private static class ClassificationAnalysis {
        int recordCount;
        double avgTotalAmount, avgOperationalEfficiency, avgRevenuePerMile;
        double avgTipRate, avgSuggestedPriceMultiplier;
        double excellentEfficiencyRatio, goodEfficiencyRatio, hotspotRatio;
        double classificationAccuracy, classificationPrecision, classificationRecall;
    }

    private static class RegressionAnalysis {
        int recordCount;
        double avgTotalAmount, avgOperationalEfficiency, avgRevenuePerMile;
        double avgTripDistance, avgTripDuration, avgWeatherImpact;
        double revenueVariance, efficiencyVariance;
        double distanceCorrelation, efficiencyCorrelation;
        double rSquared, meanAbsoluteError, rootMeanSquareError;
    }

    private static class EnsembleAnalysis {
        int recordCount;
        double avgTotalAmount, avgOperationalEfficiency, avgPredictedDemand;
        double avgSupplyDemandRatio, avgDispatchPriority, avgSuggestedPriceMultiplier;
        double optimizationSuccessRate, hotspotCoverageRate;
        double ensembleAccuracy, ensembleStability, ensembleRobustness;
        double revenueImprovement, efficiencyGain, customerSatisfaction;
    }
}
